package saramax

import (
	"encoding/json"
	"github.com/IBM/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
	"strconv"
	"time"
)

type ConsumeHandler[T any] struct {
	fn      func(msg *sarama.ConsumerMessage, t T) error
	summary *prometheus.SummaryVec
}

func NewHandler[T any](consumer string, fn func(msg *sarama.ConsumerMessage, t T) error) *ConsumeHandler[T] {
	summary := prometheus.NewSummaryVec(prometheus.SummaryOpts{ //摘要
		Namespace: "saramax",
		Subsystem: "consumer_handler",
		Name:      consumer,
		Help:      "kafka consume观测",
	}, []string{"topic", "error"})
	return &ConsumeHandler[T]{
		fn:      fn,
		summary: summary,
	}
}
func (h *ConsumeHandler[T]) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *ConsumeHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}
func (h *ConsumeHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error {
	ch := claim.Messages()
	for msg := range ch {
		h.consumeClaim(msg)
		session.MarkMessage(msg, "")
	}
	return nil
}
func (h *ConsumeHandler[T]) consumeClaim(msg *sarama.ConsumerMessage) {
	start := time.Now()
	var err error
	defer func() {
		dura := time.Since(start).Milliseconds()
		errInfo := strconv.FormatBool(err != nil)
		h.summary.WithLabelValues(msg.Topic, errInfo).Observe(float64(dura))
	}()
	var t T
	err = json.Unmarshal(msg.Value, &t)
	if err != nil {
		zap.L().Info("反序列化消息体失败:",
			zap.String("topic", msg.Topic),
			zap.Int32("partition", msg.Partition),
			zap.Int64("offset", msg.Offset),
			zap.Error(err))
		return
	}
	err = h.fn(msg, t)
	if err != nil {
		zap.L().Info("处理消息失败:",
			zap.String("topic", msg.Topic),
			zap.Int32("partition", msg.Partition),
			zap.Int64("offset", msg.Offset),
			zap.Error(err))
	}
}
